package com.myCompany.edu.realtime.app.dwd.db;

import com.myCompany.edu.realtime.app.BaseSQLApp;
import com.myCompany.edu.realtime.common.Constant;
import com.myCompany.edu.realtime.util.FlinkSinkUtil;
import com.myCompany.edu.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @ClassName：Dwd07_UserRegister
 * @Author: xxx
 * @Date: 2022/10/21 14:38
 * @Description:
 */
public class Dwd07_UserRegister extends BaseSQLApp {

    public static void main(String[] args) {
        new Dwd07_UserRegister().init(
                3005,
                2,
                "Dwd07_UserRegister"
        );
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        // 从表中读取数据
        readOdsDb(tEnv,"Dwd07_UserRegister");

        // 选取出用户表
        Table userRegister = tEnv.sqlQuery("select " +
                " `data`['id'] id, " +
                " `data`['create_time'] create_time  " +
                " from ods_db " +
                " where `database` = 'edu' " +
                " and `table` = 'user_info' " +
                " and `type` = 'insert' ");
        tEnv.createTemporaryView("user_register", userRegister);
//        tEnv.sqlQuery("select * from user_register ").execute().print();

        // 将表写入到kafka中
        tEnv.executeSql(" create table dwd_user_register( " +
                        " id  string," +
                        " create_time string  " +
                        " ) "+ SQLUtil.getKafkaSinkSQL(Constant.TOPIC_DWD_USER_REGISTER));

        // 写入到创建的动态表中
        userRegister.executeInsert("dwd_user_register");


    }
}
